/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.query;

import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import javax.jdo.identity.SingleFieldIdentity;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.datanucleus.ClassLoaderResolver;
import org.datanucleus.ObjectManager;
import org.datanucleus.exceptions.NucleusDataStoreException;
import org.datanucleus.exceptions.NucleusException;
import org.datanucleus.metadata.AbstractClassMetaData;
import org.datanucleus.metadata.DiscriminatorMetaData;
import org.datanucleus.query.evaluator.JDOQLEvaluator;
import org.datanucleus.query.evaluator.JavaQueryEvaluator;
import org.datanucleus.query.expression.Expression;
import cn.edu.thu.laud.objectstore.CassandraConsistency;
import cn.edu.thu.laud.objectstore.CassandraManagedConnection;
import cn.edu.thu.laud.objectstore.CassandraStoreManager;
import cn.edu.thu.laud.utils.serde.ConverterContext;
import cn.edu.thu.laud.objectstore.query.CassandraBooleanExpression;
import cn.edu.thu.laud.objectstore.index.IndexHandler;
import cn.edu.thu.laud.objectstore.index.IndexMeta;
import cn.edu.thu.laud.objectstore.index.KVColumn;
import cn.edu.thu.laud.utils.MetaDataUtils;
import org.datanucleus.store.query.AbstractJDOQLQuery;
import org.datanucleus.util.ClassUtils;
import org.datanucleus.util.NucleusLogger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @author zhuoan
 *
 */
public class JDOQLQuery extends AbstractJDOQLQuery {
	
	
	private static final Logger log = LoggerFactory.getLogger(JDOQLQuery.class);
	private ConverterContext converterContext;
	/**
     * Constructs a new query instance that uses the given persistence manager.
     *
     */
    public JDOQLQuery(ObjectManager om) {
        this(om, (JDOQLQuery) null);        
    }

    /**
     * Constructs a new query instance having the same criteria as the given
     * query.
     * @param q  The query from which to copy criteria.
     */
    public JDOQLQuery(ObjectManager om, JDOQLQuery q) {
        super(om, q);
        converterContext = ((CassandraStoreManager)om.getStoreManager()).getConverterContext();
    }

    /**
     * Constructor for a JDOQL query where the query is specified using the
     * "Single-String" format.
     * @param query The query string
     */
    public JDOQLQuery(ObjectManager om, String query) {
        super(om, query);
        converterContext = ((CassandraStoreManager)om.getStoreManager()).getConverterContext();
    }
    
    /**
	 * Used to load specific keys
	 * 
	 * @param ec
	 * @param candidateClass
	 * @param keys
	 * @param subclasses
	 * @param ignoreCache
	 * @param limit
	 * @param startKey
	 * @return
	 */
	public List<?> getObjectsOfCandidateType(Set<KVColumn> keys,
			AbstractClassMetaData acmd, ClassLoaderResolver clr,
			boolean subclasses, ByteBuffer identityColumn,
			ByteBuffer descriminatorColumn) {

		List<Object> results = new ArrayList<Object>(keys.size());
		// String tempKey = null;

		for (KVColumn idBytes : keys) {

			Class<?> targetClass = candidateClass;
            
			log.info("KVColumn: " + idBytes);
			
			if (descriminatorColumn != null) {

				String descriminatorValue = converterContext.getString(idBytes.getColumnValue(
						descriminatorColumn));

				String className = org.datanucleus.metadata.MetaDataUtils
						.getClassNameFromDiscriminatorValue(descriminatorValue,
								acmd.getDiscriminatorMetaData(), om);

				targetClass = clr.classForName(className);

			}

			Object identity =   
					converterContext.getObjectIdentity(om, targetClass, idBytes.getColumnValue(identityColumn));

			// Not a valid subclass, don't return it as a candidate
			if (!(identity instanceof SingleFieldIdentity)) {
				throw new NucleusDataStoreException(
						"Only single field identities are supported");
			}

			if (!ClassUtils.typesAreCompatible(candidateClass,
					((SingleFieldIdentity) identity).getTargetClassName(), clr)) {
				continue;
			}

			Object returned = om.findObject(identity, true, subclasses,
					candidateClass.getName());

			if (returned != null) {
				results.add(returned);
			}
		}

		return results;

	}

	/**
	 * Get all keys from a given column family. Used ranges to set the max
	 * amount
	 * 
	 * @param cfName
	 * @param selectColumns
	 * @return
	 * @throws Exception
	 */
	private Set<KVColumn> getAll(Cassandra.Client client, String cfName,
			List<ByteBuffer> selectColumns, int maxSize) {

		Set<KVColumn> candidateKeys = new LinkedHashSet<KVColumn>();

		KeyRange range = new KeyRange();
		range.setStart_key(ByteBufferUtil.EMPTY_BYTE_BUFFER);
		range.setEnd_key(ByteBufferUtil.EMPTY_BYTE_BUFFER);
		range.setCount(maxSize);

		List<KeySlice> results;
		SlicePredicate slice_predicate = new SlicePredicate();
		slice_predicate.setColumn_names(selectColumns);
		try {
			results = client.get_range_slices(
					new ColumnParent(cfName), slice_predicate, range, 
					CassandraConsistency.get());
		} catch (Exception e) {
			throw new NucleusException("Error scanning rows", e);
		}

		KVColumn cols;

		for (KeySlice entry : results) {

			if (entry.getColumnsSize() == 0) {
				continue;
			}

			cols = new KVColumn(entry.key);

			for (ColumnOrSuperColumn currentCol : entry.getColumns()) {

				cols.addResult(currentCol.getColumn());
			}

			candidateKeys.add(cols);
		}

		return candidateKeys;
	}

	//loads all the keys and then loads the object one by one
    @SuppressWarnings("rawtypes")
	@Override
    protected Object performExecute(Map parameters) {
          
		long startTime = System.currentTimeMillis();
        Expression exp = this.compilation.getExprFilter();
        AbstractClassMetaData acmd = om.getMetaDataManager()
                .getMetaDataForClass(candidateClass,
                        om.getClassLoaderResolver());
        
        CassandraStoreManager storeMgr = (CassandraStoreManager) om
				.getStoreManager();
        try {
           
            if (NucleusLogger.QUERY.isDebugEnabled()) {
                NucleusLogger.QUERY.debug(LOCALISER.msg("021046", "JDOQL",
                        getSingleStringQuery(), null));
            }

            CassandraManagedConnection mconn = (CassandraManagedConnection) storeMgr.getConnection(om);
            Cassandra.Client client = (Cassandra.Client)mconn.getConnection();
            String keySpace = storeMgr.getConnectionInfo().getKeyspace();
            client.set_keyspace(keySpace);
            
            int range = IndexHandler.DEFAULT_COUNT;

    		if (this.getRange() != null) {
    			range = (int) this.getRangeToExcl();

    			if (this.getOrdering() == null) {
    				throw new NucleusDataStoreException(
    						"You cannot invoke a without an ordering expression against Cassandra. Results will be randomly ordered from Cassnadra and need order to page");

    			}
    		}
           
            String columnFamily = MetaDataUtils.getColumnFamily(acmd);
            ByteBuffer idCol = MetaDataUtils.getIdentityColumn(acmd);

            Set<KVColumn> candidateKeys = null;
            DiscriminatorMetaData discriminator = null;
            ByteBuffer descriminiatorCol = null;
            
            if (acmd.hasDiscriminatorStrategy()) {
    			discriminator = acmd.getDiscriminatorMetaData();
    			descriminiatorCol = MetaDataUtils.getDiscriminatorColumnName(discriminator);    			
    		}
            
            boolean applyFilter = false;
            CassandraBooleanExpression filterExp = null;
            if(exp != null){
            	applyFilter = true;
            	 //build index.
                IndexMeta idxmeta = new IndexMeta(columnFamily, null,
        				keySpace + "_index", keySpace
        						+ "_index_entries");
                Set<ByteBuffer> disColValues = null;
                if (acmd.hasDiscriminatorStrategy()) {
                	disColValues = MetaDataUtils.getDescriminatorValues(acmd.getFullClassName(), om.getClassLoaderResolver(), om, converterContext);
                }
                QueryToCassandraMapper mapper = new QueryToCassandraMapper(converterContext, client, compilation, parameters, range, acmd, om, this, idxmeta, descriminiatorCol, disColValues);
                mapper.compileFilter();               
                filterExp = mapper.getFilterExpression();
                //TODO
                if(filterExp != null){
                	candidateKeys = filterExp.getKeys();
                	applyFilter = ! mapper.isFilterComplete();
                }
               
            }
            
            if(exp == null || filterExp == null){

                List<ByteBuffer> selectColumns = new ArrayList<ByteBuffer>();
                selectColumns.add(idCol);
                if(descriminiatorCol != null)
                	selectColumns.add(descriminiatorCol);
    			candidateKeys = getAll(client, columnFamily, selectColumns, range);
    			log.info("No index help and will scan all rows to get candidate keys.");
    		}
            
            Collection<?> results = getObjectsOfCandidateType(candidateKeys, acmd,
    				om.getClassLoaderResolver(), subclasses, idCol, descriminiatorCol);
            
            if (this.getOrdering() != null || this.getGrouping() != null || applyFilter) {

                log.warn("Cassandra unsupported expression will be evaluated in memory");
            	// Apply any result restrictions to the results
                JavaQueryEvaluator resultMapper = new JDOQLEvaluator(this,
                        results, compilation, parameters, om
                        .getClassLoaderResolver());
                results = resultMapper.execute(applyFilter, true, true, true,
                        true);
            }
            if (NucleusLogger.QUERY.isDebugEnabled()) {
                NucleusLogger.QUERY.debug(LOCALISER.msg("021074", "JDOQL", ""
                        + (System.currentTimeMillis() - startTime)));
            }

            if (type == BULK_DELETE)
            {
                om.deleteObjects(results.toArray());
                return Long.valueOf(results.size());
            }
            else if (type == BULK_UPDATE)
            {
                throw new NucleusException("Bulk Update is not yet supported");
            }
            else
            {
                return results;
            }

        } catch (Exception e) {
            throw new NucleusDataStoreException("Error on query execution",e);
        }
    }

}
